package com.atguigu.flink.day05;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author Felix
 * @date 2023/12/5
 * 该案例演示了窗口处理函数-reduce
 */
public class Flink02_window_aggregate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());


        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);

        WindowedStream<WaterSensor, String, TimeWindow> windowDS = keyedDS
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        //使用aggregate算子对窗口中的数据进行增量聚合计算
        //aggregate: 窗口中的元素类型、累加类型和聚合之后向下游传递的元素类型可以不一致
        //createAccumulator:初始化累加器~~~属于当前窗口第一个元素进来的时候执行 一个窗口只会初始化一次
        //add:当窗口中有数据来的时候，进行累加~~~只要窗口有元素进来就会进行累加
        //getResult:获取累加的结果 ~~~ 当窗口触发计算的时候调用
        //merge：会话窗口需要实现

        windowDS
            .aggregate(
                new AggregateFunction<WaterSensor, Integer, String>() {
                    @Override
                    public Integer createAccumulator() {
                        System.out.println("~~初始化累加器createAccumulator~~");
                        return 0;
                    }

                    @Override
                    public Integer add(WaterSensor value, Integer accumulator) {
                        System.out.println("~~累加add~~");
                        return accumulator + value.vc;
                    }

                    @Override
                    public String getResult(Integer accumulator) {
                        System.out.println("~~获取窗口结果getResult~~");
                        return accumulator.toString();
                    }

                    @Override
                    public Integer merge(Integer a, Integer b) {
                        System.out.println("~~merge~~");
                        return null;
                    }
                }
            ).print("~~~");

        env.execute();
    }
}
